/*

Copyright (C) 2007 Manfred Gschweidl (manfred.gschweidl@students.fhv.at)

This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA

*/

#include "GramJobCallback.h"

#include "CredManager.h"

using namespace globus_job;

namespace globus_gram_jobs {

//callback funtion registered by callback server
static void callback_func(void * user_callback_arg, char * job_contact,
		int state, int errorcode) {
	GramJobCallback
			* Monitor = (GramJobCallback*) user_callback_arg;
	GramJob* job = Monitor->getJob(job_contact);

	switch (state) {
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_STAGE_IN:
		cout << "Staging file in on: "<< job_contact << endl;
		break;
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_STAGE_OUT:
		cout << "Staging file out  on: "<< job_contact << endl;
		break;
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_PENDING:
		break; /* Reports state change to the user */

	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_ACTIVE:
		break; /* Reports state change to the user */

	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_FAILED:
		Monitor->remove(job_contact);
		job->setFailed();
		job->setDone();
		cerr << "Job Failed on: "<< job_contact << endl;
		break; /* Reports state change to the user */

	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE:
		cout << "Job Finished on: "<< job_contact << endl;
		cout << "GlobusGramJob jobcontact: "<< job->getJobContact() << endl;
		Monitor->remove(job_contact);
		job->setDone();
		break; /* Reports state change to the user */
	}
}

//callback funtion registered by job for receiving jobcontact
static void request_callback(void * user_callback_arg,
		globus_gram_protocol_error_t failure_code, const char * job_contact,
		globus_gram_protocol_job_state_t state,
		globus_gram_protocol_error_t errorcode) {
	GramJob* Request = (GramJob*) user_callback_arg;
	
	if (job_contact != NULL) {
		cout << strlen(job_contact) << endl;	
	} else {
		cout << "Error during the code submission"<< endl << "Error Code:"
						<< failure_code << endl;
		Request->setDone();
		Request->setFailed();
		return;
	}
	

	if (strlen(job_contact) == 0) {
		cout << "Error during the code submission"<< endl << "Error Code:"
				<< failure_code << endl;
		Request->setDone();
		Request->setFailed();
		return;
	}

	cout << "Contact on the server: "<< job_contact << endl;

	if (failure_code==0) {

		Request->setJobContact(const_cast<char*>(job_contact));
		if (Request->onlyGetJobHandle()) {
			Request->setDone();
		}
	} else {
		cout << "Error during the code submission"<< endl << "Error Code:"
				<< failure_code << endl;
		Request->setDone();
		Request->setFailed();
	}
}

}

namespace globus_job {

static void initModules();
static void destroyModules();

void initModules() {
	globus_module_activate(GLOBUS_COMMON_MODULE);
	globus_module_activate(GLOBUS_GRAM_CLIENT_MODULE);
}

void destroyModules() {
	globus_module_deactivate(GLOBUS_GRAM_CLIENT_MODULE);
	globus_module_deactivate(GLOBUS_COMMON_MODULE);
}

GramJob::GramJob(GramJobCallback* f) :
	failed(false), jobcontact(NULL), callback(f), only_get_jobHandle(true) {
	initModules();
	globus_mutex_init(&jobcontactmutex, GLOBUS_NULL);
	globus_gram_client_attr_init (&attr);

	globus_mutex_init(&mutex, GLOBUS_NULL);
	globus_cond_init(&cond, GLOBUS_NULL);
	done = GLOBUS_FALSE;
}

GramJob::GramJob(GramJobCallback* f, string jc) :
	failed(false), callback(f), only_get_jobHandle(true) {
	initModules();
	//need to copy string because globus resuses memroy allocted by pointer
	jobcontact = new char[strlen(jc.c_str())+1];
	char* temp=strcpy(jobcontact, jc.c_str());
	globus_mutex_init(&jobcontactmutex, GLOBUS_NULL);
	globus_gram_client_attr_init (&attr);

	globus_mutex_init(&mutex, GLOBUS_NULL);
	globus_cond_init(&cond, GLOBUS_NULL);
	done = GLOBUS_FALSE;
}

GramJob::GramJob(string jc) :
	failed(false), callback(NULL), only_get_jobHandle(true) {
	initModules();
	//need to copy string because globus resuses memroy allocted by pointer
	jobcontact = new char[strlen(jc.c_str())+1];
	char* temp=strcpy(jobcontact, jc.c_str());
	globus_mutex_init(&jobcontactmutex, GLOBUS_NULL);
	globus_gram_client_attr_init (&attr);

	globus_mutex_init(&mutex, GLOBUS_NULL);
	globus_cond_init(&cond, GLOBUS_NULL);
	done = GLOBUS_FALSE;
}

GramJob::~GramJob() {
	if (jobcontact!=NULL) {
		globus_gram_client_job_contact_free(jobcontact);
		jobcontact=NULL;
	}
	globus_gram_client_attr_destroy(&attr);
	globus_mutex_destroy(&jobcontactmutex);
	destroyModules();

	globus_mutex_destroy(&mutex);
	globus_cond_destroy(&cond);
}

//get job status from globus - blocking implementation
int GramJob::getJobStatus() {
	if (jobcontact == NULL) {
		cout << "no jobcontact to query status"<< endl;
		return -1;
	}
	int state = 0;
	int failureCode = 0;
	int rc = globus_gram_client_job_status(jobcontact, &state, &failureCode);

	if (rc != GLOBUS_SUCCESS) {
		// assume job is finished ;)
		return -1;
	}

	switch (state) {
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_PENDING:
		rc = 1;
		break; /* Reports state change to the user */
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_ACTIVE:
		rc = 2;
		break; /* Reports state change to the user */
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_FAILED:
		if (callback != NULL) {
			callback->remove(jobcontact);
		}
		setFailed();
		setDone();
		cerr << "Job Failed on: "<< jobcontact << endl;
		rc = 4;
		break; /* Reports state change to the user */
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE:
		cout << "Job Finished on: "<< jobcontact << endl;
		if (callback != NULL) {
			callback->remove(jobcontact);
		}
		rc = 3;
		setDone();
		break; /* Reports state change to the user */
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_SUSPENDED:
		rc = 5;
		break; /* Reports state change to the user */
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_UNSUBMITTED:
		rc = 6;
		break; /* Reports state change to the user */
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_STAGE_IN:
		cout << "Staging file in on: "<< jobcontact << endl;
		rc = 7;
		break;
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_STAGE_OUT:
		cout << "Staging file out  on: "<< jobcontact << endl;
		rc = 8;
		break;
	default:
		rc = -1;
	}
	return state;
}

string GramJob::printJobStatus(int statusCode) {

	string userCode = "Unknown";

	switch (statusCode) {
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_PENDING:
		userCode = "Waiting";
		break;
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_ACTIVE:
		userCode = "Active";
		break;
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_FAILED:
		userCode = "Failed";
		break;
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE:
		userCode = "Finished";
		break;
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_SUSPENDED:
		userCode = "Suspended";
		break;
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_UNSUBMITTED:
		userCode = "Unsubmitted";
		break;
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_STAGE_IN:
		userCode = "Staging in";
		break;
	case GLOBUS_GRAM_PROTOCOL_JOB_STATE_STAGE_OUT:
		userCode = "Staging out";
		break;
	default:
		break;
	}
	return userCode;
}

bool GramJob::onlyGetJobHandle() {
	return only_get_jobHandle;
}

void GramJob::setOnlyGetJobHandle(bool state) {
	only_get_jobHandle = state;
}

//submit callback function sets jobcontact
void GramJob::setJobContact(const char* c) {
	//need to copy again string because of pointer change by globus
	jobcontact = new char[strlen(c)+1];
	char* temp=strcpy(jobcontact, c);
	callback->add(string(c), this);
}

//submit job to globus and register callback funtion to receive jobcontact
bool GramJob::submit(string res, string rsl) {
	failed=false;

	//globus_gram_client_attr_t attr;
	globus_gram_client_attr_init (&attr);

	CredManager::acquireCredential(credential);

	// redirect the credentials to the GRAM client
	int rc = globus_gram_client_set_credentials(credential);
	
	if (rc != 0) {
		cout << "gram error: could not set credentials"<< endl;
		failed = true;
		return false;
	}

	
	rc = globus_gram_client_attr_set_credential (attr, credential);

	if (rc != 0) {
		cout << "gram error: could not set credentials"<< endl;
		failed = true;
		return false;	
	}

	rc = globus_gram_client_register_job_request(res.c_str(), rsl.c_str(),
			GLOBUS_GRAM_PROTOCOL_JOB_STATE_ALL, callback->getURL(), attr,
			globus_gram_jobs::request_callback, (void*) this);

	if (rc != 0) /* if there is an error */
	{
		printf("gram error: %d - %s\n", rc,
		/* translate the error into english */
		globus_gram_client_error_string(rc));
		failed=true;
		return false;
	}

	return true;
}

globus_bool_t GramJob::isDone() {
	return done;
}

//possible reuse of object for further job submissions by callback server
void GramJob::proceed() {
	globus_mutex_lock(&jobcontactmutex);
	/* Free up the resources of the job_contact, as the job is over, and
	 * the contact is now useless.
	 */
	if (jobcontact!=NULL) {
		globus_gram_client_job_contact_free(jobcontact);
		jobcontact=NULL;
	}

	globus_mutex_lock(&mutex);
	done = GLOBUS_FALSE;
	globus_mutex_unlock(&mutex);

	//GlobusCallback::proceed();
	globus_mutex_unlock(&jobcontactmutex);
}

void GramJob::wait() {
	globus_mutex_lock(&mutex);
	while (!isDone()) {
		globus_cond_wait(&cond, &mutex);
	}
	globus_mutex_unlock(&mutex);
}

void GramJob::setDone() {
	globus_mutex_lock(&mutex);
	done = GLOBUS_TRUE;
	globus_cond_broadcast(&cond);
	globus_mutex_unlock(&mutex);
}

//cancle job - blocking implementation
bool GramJob::cancel() {
	if (jobcontact == NULL) {
		cout << "no jobcontact to query status"<< endl;
		return false;
	}
	int rc;
	printf("\tsending cancel to job manager...\n");

	if ((rc = globus_gram_client_job_cancel(jobcontact)) != GLOBUS_SUCCESS) {
		printf("\tFailed to cancel job.\n");
		printf("\tgram error: %d - %s\n", rc,
				globus_gram_client_error_string(rc));
		return false;
	} else {
		printf("\tjob cancel was successful.\n");
		return true;
	}
}

void GramJob::setFailed() {
	failed=true;
}

bool GramJob::hasFailed() {
	return failed;
}

string GramJob::getJobContact() {
	if (jobcontact != NULL) {
		return string(jobcontact);
	}
	cout << "Job Contact is NULL!" << endl;
	return "";
}

bool GramJob::ping(URL* host) {

	bool success = true;
	gss_cred_id_t credential;

	// initialize GSS
	if ( !CredManager::acquireCredential(credential) ) {
		return false;
	}

	int rc = globus_gram_client_ping(host->getHost().c_str());

	if (rc != GLOBUS_SUCCESS ) {
		success = false;
	}

	if ( !CredManager::releaseCredential(credential) ) {
		return false;
	}

	return success;
}

GramJobCallback::GramJobCallback() {
	globus_module_activate(GLOBUS_GRAM_CLIENT_MODULE);
	globus_mutex_init(&JobsTableMutex, GLOBUS_NULL);
	//register callback funtion for job status changes
	globus_gram_client_callback_allow(globus_gram_jobs::callback_func,
			(void *) this, &callback_contact);
	cout << "Gram contact " << callback_contact << endl;
}

char* GramJobCallback::getURL() {
	return callback_contact;
}

GramJob* GramJobCallback::getJob(char* s) {
	return JobsTable[s];
}

GramJobCallback::~GramJobCallback() {
	//cout << callback_contact << " destroyed" << endl;
	//deregister callback function
	globus_gram_client_callback_disallow(callback_contact);
	globus_free(callback_contact);
	globus_mutex_destroy(&JobsTableMutex);
	globus_module_activate(GLOBUS_GRAM_CLIENT_MODULE);
}

//add job to internal map
void GramJobCallback::add(string jobcontact, GramJob* job) {
	lock();
	JobsTable[jobcontact]=job;
	unLock();
}

//remove job form itnernal map identified by jobcontact
void GramJobCallback::remove(char* jobcontact) {
	lock();
	JobsTable.erase(jobcontact);
	unLock();
}

void GramJobCallback::lock() {
	globus_mutex_lock(&JobsTableMutex);
}

void GramJobCallback::unLock() {
	globus_mutex_unlock(&JobsTableMutex);
}

}
